package com.ffcs.crmd.tsp.server.manager;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ffcs.crmd.tsp.api.dto.client.TspJobDetailThreadInfo;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.core.util.RemotingUtil;
import com.ffcs.crmd.tsp.server.application.ApplicationManagerHolder;
import com.ffcs.crmd.tsp.server.core.DataBaseHandler;
import com.ffcs.crmd.tsp.server.core.RemotingHandler;
import com.ffcs.crmd.tsp.server.jdbc.entity.Job;
import com.ffcs.crmd.tsp.server.job.info.JobDetailShell;
import com.ffcs.crmd.tsp.server.netty.task.TaskRegisterInfo;
import com.ffcs.crmd.tsp.server.util.PingHelper;

import io.netty.channel.Channel;

/**
 * 
 * 功能说明:执行端通信管理器
 *
 * @author ZHONGFUHUA
 * 
 * @since 2.0.0-SNAPSHOT
 *
 */
public class TaskRemotingManager {
    
    /**
     * 日志
     */
    private static final Logger                     LOG               = LoggerFactory.getLogger(LoggerName.RemotingLoggerName);
    
    /**
     * 通道失效超时时间
     */
    private final static long                       channelTimeout    = 120 * 1000;
    
    /**
     * 通信处理中心
     */
    private RemotingHandler                         remotingHandler;
    
    
    /**
     * 数据源处理中心
     */
    private DataBaseHandler                         dataBaseHandler;
    
    /**
     * 通道表
     */
    private volatile Map<String, Vector<Channel>>   identityTable     = new ConcurrentHashMap<String, Vector<Channel>>();
    
    /**
     * 通道表
     */
    private volatile Map<Channel, TaskRegisterInfo> channelTable      = new ConcurrentHashMap<Channel, TaskRegisterInfo>();
    
    /**
     * 
     */
    private volatile Map<String, Channel>           chennelOfflineMap = new ConcurrentHashMap<String, Channel>();
    
    /**
     * 任务端管理器
     */
    private JobProcessManager                       jobProcessManager = ApplicationManagerHolder.getBean("jobProcessManager",
        JobProcessManager.class);
    
    /**
     * 
     * 功能说明:注册执行端
     * 
     * @param channel 通道
     * @param taskRegisterInfo 执行端注册信息
     */
    public void registerTask(Channel channel, TaskRegisterInfo taskRegisterInfo) {
        boolean exist = false;
        for (Channel channelCache : channelTable.keySet()) {
            if (RemotingUtil.getChannelId(channel).equals(RemotingUtil.getChannelId(channelCache)) || channel == channelCache) {
                channelTable.put(channelCache, taskRegisterInfo);
                exist = true;
            }
        }
        
        
        
        if (!exist) {
            LOG.info("SERVER-TASK: new channel register success - clientId[{}] identity[{}] channel[{}]", taskRegisterInfo.getClientId(),
                taskRegisterInfo.getIdentity(), channel);
            putChannelToJobProcess(taskRegisterInfo.getIdentity(), channel);
            if (chennelOfflineMap.get(taskRegisterInfo.getClientId()) != null) {
                Channel channelOffline = chennelOfflineMap.remove(taskRegisterInfo.getClientId());
                List<JobDetailShell> jobDetailShellList = jobProcessManager.getJobDetailOffline(channelOffline);
                LOG.debug("SERVER-TASK: restoring the offline job detail - clientId[{}] size[{}] channel[{}]", taskRegisterInfo.getClientId(),
                    jobDetailShellList.size(), channelOffline);
                for (JobDetailShell jobDetailShellOffline : jobDetailShellList) {
                    TspJobDetailThreadInfo tspJobDetailThreadInfo = remotingHandler.queryThreadInfo(jobDetailShellOffline, channel);
                    if (tspJobDetailThreadInfo != null && tspJobDetailThreadInfo.getJobProcessThread() == null) {
                        jobProcessManager.completeJobDetail(jobDetailShellOffline.getJobName(), channelOffline,
                            jobDetailShellOffline.getJobDetailName(), jobDetailShellOffline.getId());
                        LOG.warn("SERVER-TASK: restoring the offline job detail[{}] to complete. id=[{}] opaque=[{}]", jobDetailShellOffline.getJobDetailName(),
                            jobDetailShellOffline.getId(), jobDetailShellOffline.getRemotingOpaque());
                    } else {
                        jobDetailShellOffline.updateOnline();
                        jobProcessManager.removeJobDetailRunning(channelOffline, jobDetailShellOffline);
                        jobProcessManager.putJobDetailRunning(jobDetailShellOffline, channel);
                        LOG.warn("SERVER-TASK: restoring the offline job detail[{}] to online. id=[{}] opaque=[{}]", jobDetailShellOffline.getJobDetailName(),
                            jobDetailShellOffline.getId(), jobDetailShellOffline.getRemotingOpaque());
                    }
                }
            }
            channelTable.put(channel, taskRegisterInfo);
            putIdentityTable(channel, taskRegisterInfo);
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param taskRegisterInfo
     */
    private void putIdentityTable(Channel channel, TaskRegisterInfo taskRegisterInfo) {
        Vector<Channel> channelList = this.identityTable.get(taskRegisterInfo.getIdentity());
        if (CollectionUtils.isEmpty(channelList)) {
            synchronized (this) {
                channelList = this.identityTable.get(taskRegisterInfo.getIdentity());
                if (CollectionUtils.isEmpty(channelList)) {
                    channelList = new Vector<Channel>();
                    this.identityTable.put(taskRegisterInfo.getIdentity(), channelList);
                }
            }
        }
        
        if (!channelList.contains(channel)) {
            channelList.add(channel);
        }
        
        this.identityTable.put(taskRegisterInfo.getIdentity(), channelList);
    }
    
    /**
     * 
     * 功能说明:扫描定时没有更新的通道，并关闭移除
     *
     */
    public void scanChannelTimeout() {
        Iterator<Entry<Channel, TaskRegisterInfo>> channelTableIt = channelTable.entrySet().iterator();
        while (channelTableIt.hasNext()) {
            Entry<Channel, TaskRegisterInfo> entryTable = channelTableIt.next();
            Channel channel = entryTable.getKey();
            TaskRegisterInfo taskRegisterInfo = entryTable.getValue();
            long diff = System.currentTimeMillis() - taskRegisterInfo.getLastUpdateTimestamp();
            if (diff >= channelTimeout) {
                if(!remotingHandler.testConnect(channel)){
                    jobProcessManager.removeChannelJobAndOffline(channel);
                    identityTable.get(taskRegisterInfo.getIdentity()).remove(channel);
                    chennelOfflineMap.put(taskRegisterInfo.getClientId(), channel);
                    channelTableIt.remove();
                    RemotingUtil.closeChannel(channel);
                    LOG.warn("SERVER-TASK: scan timeout channel will to close - clientId[{}] channel[{}]", taskRegisterInfo.getClientId(), channel);
                }else{
                    LOG.warn("SERVER-TASK: scan timeout channel but successful communication - clientId[{}] channel[{}]", taskRegisterInfo.getClientId(), channel);
                }
            }
        }
    }
    
    /**
     * 
     * 功能说明:执行关闭通道事件
     * 
     * @param channel 通道
     */
    public void channelCloseEvent(Channel channel) {
        boolean exist = false;
        Iterator<Entry<Channel, TaskRegisterInfo>> channelTableIt = channelTable.entrySet().iterator();
        while (channelTableIt.hasNext()) {
            Entry<Channel, TaskRegisterInfo> entryTable = channelTableIt.next();
            Channel channelCache = entryTable.getKey();
            TaskRegisterInfo taskRegisterInfo = entryTable.getValue();
            if (RemotingUtil.getChannelId(channel).equals(RemotingUtil.getChannelId(channelCache)) || channel == channelCache) {
                identityTable.get(taskRegisterInfo.getIdentity()).remove(channelCache);
                boolean ping = PingHelper.ping(RemotingUtil.parseChannelRemote2Addr(channel), 3, 5);
                if (ping) {
                    jobProcessManager.removeChannelJobAndShutdown(channelCache);
                    LOG.info("SERVER-TASK: one channel close sucess online - channel[{}] taskInfo[{}]", channelCache, taskRegisterInfo);
                } else {
                    jobProcessManager.removeChannelJobAndOffline(channelCache);
                    chennelOfflineMap.put(taskRegisterInfo.getClientId(), channelCache);
                    LOG.info("SERVER-TASK: one channel close sucess offline - channel[{}] taskInfo[{}]", channelCache, taskRegisterInfo);
                }
                channelTableIt.remove();
                exist = true;
                break;
            }
        }
        
        if (!exist) {
            boolean ping = PingHelper.ping(RemotingUtil.parseChannelRemote2Addr(channel), 3, 5);
            if (ping) {
                jobProcessManager.removeChannelJobAndShutdown(channel);
            } 
            LOG.warn("SERVER-TASK: one channel close sucess but may has been remove - channel[{}]", channel);
        }
    }
    
    /**
     * 
     * 功能说明:更新内存中执行端关联的任务信息
     *
     */
    private void putChannelToJobProcess(String identity, Channel channel) {
        try {
            Job jobParam = new Job();
            jobParam.setIdentity(identity);
            List<Job> jobList = dataBaseHandler.getServiceSupport().queryJob(jobParam);
            for(Job job : jobList){
                jobProcessManager.putChannelJob(job.getName(), channel);
            }
        } catch (Exception e) {
            LOG.error("SERVER-SCHEDULER: put channel to JobProcess exception:", e);
        }
    }
    
    
    /**
     * 
     * 功能说明:获取执行端标识
     * 
     * @return 执行端标识
     */
    public String getIdentity(Channel channel) {
        TaskRegisterInfo taskRegisterInfo = channelTable.get(channel);
        if (taskRegisterInfo != null) {
            return taskRegisterInfo.getIdentity();
        }
        return null;
    }
    
    /**
     * 功能说明:获取通道表
     * 
     * @return 通道表
     */
    public Map<Channel, TaskRegisterInfo> getChannelTable() {
        Map<Channel, TaskRegisterInfo> map = new HashMap<Channel, TaskRegisterInfo>();
        map.putAll(channelTable);
        return map;
    }
    
    /**
     * 功能说明:获取注册信息
     * 
     * @return 通道表
     */
    public TaskRegisterInfo getTaskRegisterInfo(Channel channel) {
        return channelTable.get(channel);
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param identity
     * @return
     */
    public Vector<Channel> getChannels(String identity) {
        return this.identityTable.get(identity);
    }
    
    /**
     * 功能说明: 获取任务端管理器
     *
     * @return jobProcessManager 任务端管理器
     */
    public JobProcessManager getJobProcessManager() {
        return jobProcessManager;
    }
    
    /**
     * 功能说明: 设置任务端管理器
     *
     * @param jobProcessManager 任务端管理器 
     */
    public void setJobProcessManager(JobProcessManager jobProcessManager) {
        this.jobProcessManager = jobProcessManager;
    }
    
    /**
     * 功能说明:
     * 
     * @param remotingHandler
     */
    public void setRemotingHandler(RemotingHandler remotingHandler) {
        this.remotingHandler = remotingHandler;
    }
    
    /**
     * 功能说明: 设置数据源处理中心
     *
     * @param dataBaseHandler 数据源处理中心 
     */
    public void setDataBaseHandler(DataBaseHandler dataBaseHandler) {
        this.dataBaseHandler = dataBaseHandler;
    }

    /**
     * 功能说明:
     * 
     * @param identity
     */
    public void updateJobChannel(String jobName, String identity) {
        try {
            Vector<Channel> channels = this.identityTable.get(identity);
            for(Channel channel : channels){
                jobProcessManager.putChannelJob(jobName, channel);
            }
        } catch (Exception e) {
            LOG.error("SERVER-SCHEDULER: put channel to JobProcess exception:", e);
        }
    }

    
    /**
     * 功能说明:
     * 
     * @param identity
     */
    public void deleteJobChannel(String jobName, String identity) {
        try {
            Vector<Channel> channels = this.identityTable.get(identity);
            for(Channel channel : channels){
                jobProcessManager.removeChannelJob(jobName, channel);
            }
        } catch (Exception e) {
            LOG.error("SERVER-SCHEDULER: put channel to JobProcess exception:", e);
        }
    }
}
